package com.tiancy.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * TODO 使用 lambda 简化代码 注意事项
 * 流处理写法统计无界流中数据统计单词频次 . 用 DataStream API做了批处理的实现
 *  基本思路：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。
 *  在 hadoop202 上执行 nc -lk 7777
 *  在集群上执行 wordCount 案例
 *  - 1、将当前项目打包,并通过 hadoop202:8081 将当前项目的jar 上传到当前集群中 .
 *  - 2、通过 webUI 上传 jar 后,需要点击上传的jar,然后将当前文件的全类名放上去: com.tiancy.wc.Using_4_SocketWordCount_LambdaNote
 */
public class Using_4_SocketWordCount_LambdaNote {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> socketLineDs = env.socketTextStream("hadoop202", 7777);
        /*
            TODO 使用 lambda 语法简化代码在JVM执行时会进行类型擦除,具体报错如下:
                    Caused by: org.apache.flink.api.common.functions.InvalidTypesException: The generic type parameters of 'Collector' are missing.
                    In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved.
                    An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.api.common.functions.FlatMapFunction' interface.
                    Otherwise the type has to be specified explicitly using type information.
                缺少“Collector”的泛型类型参数。在许多情况下，当涉及Java泛型时，lambda方法不能为自动类型提取提供足够的信息。 一个简单的解决方法是使用一个（匿名）类来实现“org.apache.flink.api.common.functions.FlatMapFunction”接口。否则，必须使用类型信息显式指定类型。
                    Java的类型擦除。在Java中，泛型信息在编译时期就被擦除了，这意味着在运行时，JVM实际上并不知道泛型的具体类型。这就是为什么你会在运行这段代码时遇到问题。
                    在Flink中，算子如flatMap需要知道它的输入和输出类型，以便在网络中进行序列化和反序列化，同时也用于内存和磁盘之间的数据交换。因此，当类型信息被擦除时，Flink就无法获取到正确的类型信息。
                    解决方法是为Flink提供明确的类型信息。在Flink中，我们可以通过实现ResultTypeQueryable接口或者提供一个显式的TypeInformation来做到这一点。

         */
        // 下面这段写法会报错,报错的原因就是:使用 lambda 语法在编译期间会进行类型擦除.
          /*  socketLineDs.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
                String[] words = value.split(" ");
                for (String word : words) {
                    Tuple2<String, Integer> wordToOne = Tuple2.of(word, 1);
                    out.collect(wordToOne);
                }
            }).keyBy(0).sum(1).print();*/

        // 正确写法如下
        socketLineDs.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (value, out) -> {
                    String[] words = value.split(" ");
                    for (String word : words) {
                        Tuple2<String, Integer> wordToOne = Tuple2.of(word, 1);
                        out.collect(wordToOne);
                    }
                })
                // 显示指定当前算子的返回值类型,写法相对固定:Types
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0).sum(1).print();
        env.execute();
    }
}